package com.atguigu.hadoop.examples.mapreduce;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Random;

public class QuasiMonteCarlo extends Configured
        implements Tool
{
    static final String DESCRIPTION = "A map/reduce program that estimates Pi using a quasi-Monte Carlo method.";
    private static final String TMP_DIR_PREFIX = QuasiMonteCarlo.class.getSimpleName();

    public static BigDecimal estimatePi(int numMaps, long numPoints, Path tmpDir, Configuration conf)
            throws IOException, ClassNotFoundException, InterruptedException
    {
        Job job = Job.getInstance(conf);

        job.setJobName(QuasiMonteCarlo.class.getSimpleName());
        job.setJarByClass(QuasiMonteCarlo.class);

        job.setInputFormatClass(SequenceFileInputFormat.class);

        job.setOutputKeyClass(BooleanWritable.class);
        job.setOutputValueClass(LongWritable.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

        job.setMapperClass(QmcMapper.class);

        job.setReducerClass(QmcReducer.class);
        job.setNumReduceTasks(1);

        job.setSpeculativeExecution(false);

        Path inDir = new Path(tmpDir, "in");
        Path outDir = new Path(tmpDir, "out");
        FileInputFormat.setInputPaths(job, new Path[] { inDir });
        FileOutputFormat.setOutputPath(job, outDir);

        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(tmpDir)) {
            throw new IOException("Tmp directory " + fs.makeQualified(tmpDir) + " already exists.  Please remove it first.");
        }

        if (!fs.mkdirs(inDir)) {
            throw new IOException("Cannot create input directory " + inDir);
        }

        try
        {
            for (int i = 0; i < numMaps; i++) {
                Path file = new Path(inDir, "part" + i);
                LongWritable offset = new LongWritable(i * numPoints);
                LongWritable size = new LongWritable(numPoints);
                SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, file, LongWritable.class, LongWritable.class, SequenceFile.CompressionType.NONE);
                try
                {
                    writer.append(offset, size);
                } finally {
                    writer.close();
                }
                System.out.println("Wrote input for Map #" + i);
            }

            System.out.println("Starting Job");
            long startTime = Time.monotonicNow();
            job.waitForCompletion(true);
            if (!job.isSuccessful()) {
                System.out.println("Job " + job.getJobID() + " failed!");
                System.exit(1);
            }
            double duration = (Time.monotonicNow() - startTime) / 1000.0D;
            System.out.println("Job Finished in " + duration + " seconds");

            Path inFile = new Path(outDir, "reduce-out");
            LongWritable numInside = new LongWritable();
            LongWritable numOutside = new LongWritable();
            SequenceFile.Reader reader = new SequenceFile.Reader(fs, inFile, conf);
            try {
            }
            finally {
                reader.close();
            }

            BigDecimal numTotal = BigDecimal.valueOf(numMaps)
                    .multiply(BigDecimal.valueOf(numPoints));
            return BigDecimal.valueOf(4L).setScale(20)
                    .multiply(BigDecimal.valueOf(numInside
                            .get()))
                    .divide(numTotal, RoundingMode.HALF_UP);
        }
        finally {
            fs.delete(tmpDir, true);
        }
    }

    public int run(String[] args)
            throws Exception
    {
        if (args.length != 2) {
            System.err.println("Usage: " + getClass().getName() + " <nMaps> <nSamples>");
            ToolRunner.printGenericCommandUsage(System.err);
            return 2;
        }

        int nMaps = Integer.parseInt(args[0]);
        long nSamples = Long.parseLong(args[1]);
        long now = System.currentTimeMillis();
        int rand = new Random().nextInt(2147483647);
        Path tmpDir = new Path(TMP_DIR_PREFIX + "_" + now + "_" + rand);

        System.out.println("Number of Maps  = " + nMaps);
        System.out.println("Samples per Map = " + nSamples);

        System.out.println("Estimated value of Pi is " +
                estimatePi(nMaps, nSamples, tmpDir,
                        getConf()));
        return 0;
    }

    public static void main(String[] argv)
            throws Exception
    {
        System.exit(ToolRunner.run(null, new QuasiMonteCarlo(), argv));
    }

    public static class QmcReducer extends Reducer<BooleanWritable, LongWritable, WritableComparable<?>, Writable>
    {
        private long numInside = 0L;
        private long numOutside = 0L;

        public void reduce(BooleanWritable isInside, Iterable<LongWritable> values, Reducer<BooleanWritable, LongWritable, WritableComparable<?>, Writable>.Context context)
                throws IOException, InterruptedException
        {
            if (isInside.get()) {
                for (LongWritable val : values)
                    this.numInside += val.get();
            }
            else
                for (LongWritable val : values)
                    this.numOutside += val.get();
        }

        public void cleanup(Reducer<BooleanWritable, LongWritable, WritableComparable<?>, Writable>.Context context)
                throws IOException
        {
            Configuration conf = context.getConfiguration();
            Path outDir = new Path(conf.get("mapreduce.output.fileoutputformat.outputdir"));
            Path outFile = new Path(outDir, "reduce-out");
            FileSystem fileSys = FileSystem.get(conf);
            SequenceFile.Writer writer = SequenceFile.createWriter(fileSys, conf, outFile, LongWritable.class, LongWritable.class, SequenceFile.CompressionType.NONE);

            writer.append(new LongWritable(this.numInside), new LongWritable(this.numOutside));
            writer.close();
        }
    }

    public static class QmcMapper extends Mapper<LongWritable, LongWritable, BooleanWritable, LongWritable>
    {
        public void map(LongWritable offset, LongWritable size, Mapper<LongWritable, LongWritable, BooleanWritable, LongWritable>.Context context)
                throws IOException, InterruptedException
        {
            QuasiMonteCarlo.HaltonSequence haltonsequence = new QuasiMonteCarlo.HaltonSequence(offset.get());
            long numInside = 0L;
            long numOutside = 0L;

            for (long i = 0L; i < size.get(); )
            {
                double[] point = haltonsequence.nextPoint();

                double x = point[0] - 0.5D;
                double y = point[1] - 0.5D;
                if (x * x + y * y > 0.25D)
                    numOutside += 1L;
                else {
                    numInside += 1L;
                }

                i += 1L;
                if (i % 1000L == 0L) {
                    context.setStatus("Generated " + i + " samples.");
                }

            }

            context.write(new BooleanWritable(true), new LongWritable(numInside));
            context.write(new BooleanWritable(false), new LongWritable(numOutside));
        }
    }

    private static class HaltonSequence
    {
        static final int[] P = { 2, 3 };

        static final int[] K = { 63, 40 };
        private long index;
        private double[] x;
        private double[][] q;
        private int[][] d;

        HaltonSequence(long startindex)
        {
            this.index = startindex;
            this.x = new double[K.length];
            this.q = new double[K.length][];
            this.d = new int[K.length][];
            for (int i = 0; i < K.length; i++) {
                this.q[i] = new double[K[i]];
                this.d[i] = new int[K[i]];
            }

            for (int i = 0; i < K.length; i++) {
                long k = this.index;
                this.x[i] = 0.0D;

                for (int j = 0; j < K[i]; j++) {
                    this.q[i][j] = ((j == 0 ? 1.0D : this.q[i][(j - 1)]) / P[i]);
                    this.d[i][j] = ((int)(k % P[i]));
                    k = (k - this.d[i][j]) / P[i];
                    this.x[i] += this.d[i][j] * this.q[i][j];
                }
            }
        }

        double[] nextPoint()
        {
            this.index += 1L;
            for (int i = 0; i < K.length; i++) {
                for (int j = 0; j < K[i]; j++) {
                    this.d[i][j] += 1;
                    this.x[i] += this.q[i][j];
                    if (this.d[i][j] < P[i]) {
                        break;
                    }
                    this.d[i][j] = 0;
                    this.x[i] -= (j == 0 ? 1.0D : this.q[i][(j - 1)]);
                }
            }
            return this.x;
        }
    }
}